package com.mine.eureka.controllers;


import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.lang.reflect.Method;

@Configuration
public class ConsumerConfiguration {
    @Value("${rocketmq.name-server}")
    public String nameserver;

    @Value("${rocketmq.producer.group}")
    public String group;


    @Bean
    public void createConsumer() {
        Class<?>[] clazzes = MyConsumer.class.getDeclaredClasses();
        for (Class<?> clazz : clazzes) {
            DefaultMQPushConsumer pushConsumer = new DefaultMQPushConsumer(group);
            pushConsumer.setNamesrvAddr(nameserver);
            //集群模式
            pushConsumer.setMessageModel(MessageModel.CLUSTERING);
            //  订阅的topic tag
            try {
                Method[] methods = clazz.getMethods();
                MqAnnotation mqAnnotation = null;
                for (Method method : methods) {
                    if (method.getName().equals("consumeMessage")) {
                        mqAnnotation = method.getAnnotation(MqAnnotation.class);
                    }
                    ;
                }
                //          MqAnnotation mqAnnotation = clazz.getMethod("consumeMessage").getAnnotation(MqAnnotation.class);
                pushConsumer.subscribe(mqAnnotation.topic(), "*");
            } catch (MQClientException e) {
                throw new RuntimeException(e);
            }

            try {
                pushConsumer.registerMessageListener((MessageListenerConcurrently) clazz.newInstance());
            } catch (InstantiationException e) {
                throw new RuntimeException(e);
            } catch (IllegalAccessException e) {
                throw new RuntimeException(e);
            }


            try {
                pushConsumer.start();
            } catch (MQClientException e) {
                throw new RuntimeException(e);
            }
            System.out.printf("Broadcast Consumer Started.%n");

        }
    }

}
